package leung.java.disruptor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.junit.Test;

import com.lmax.disruptor.FatalExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.WorkerPool;

public class WorkTest {

	private static final int NUM_WORKERS = 3;
    private static final int BUFFER_SIZE = 1024 * 8;
    private static final long ITERATIONS =  100L;
    private final ExecutorService EXECUTOR = Executors.newFixedThreadPool(NUM_WORKERS);
	
	private final RingBuffer<ValueEvent> ringBuffer = 
			new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, BUFFER_SIZE);
    
	private final MyWorkHandler handlers []=new MyWorkHandler[NUM_WORKERS];
	{
		for(int i=0;i<NUM_WORKERS;i++){
			handlers[i]=new MyWorkHandler();
		}
	}
	
	private final WorkerPool<ValueEvent> workerPool = 
            new WorkerPool<ValueEvent>(ringBuffer,
                                       ringBuffer.newBarrier(),
                                       new FatalExceptionHandler(),
                                       handlers);
	
	
	@Test
	public void test(){
		
		RingBuffer<ValueEvent>  ring=workerPool.start(EXECUTOR);
		ring.setGatingSequences(workerPool.getWorkerSequences());
		
		System.out.println("ss");
		for(int i=0;i<ITERATIONS;i++){
			long index=ring.next();
			ring.get(index).setValue(i);
			ring.publish(index);
		}
		
		System.out.println("ssss");

	}

}

class MyWorkHandler  implements WorkHandler<ValueEvent>{
	@Override
	public void onEvent(leung.java.disruptor.ValueEvent event) throws Exception {
		System.out.println(Thread.currentThread().getName()+" "+event.getValue());
	}
}
